package org.ricks.net.http.client;


import org.ricks.net.extension.plugins.Plugin;
import org.ricks.net.extension.plugins.SslPlugin;
import org.ricks.net.extension.ssl.factory.ClientSSLContextFactory;
import org.ricks.net.http.client.impl.WebSocketRequestImpl;
import org.ricks.net.http.client.impl.WebSocketResponseImpl;
import org.ricks.net.http.codec.CloseReason;
import org.ricks.net.http.codec.Decoder;
import org.ricks.net.http.codec.WebSocket;
import org.ricks.net.http.enums.*;
import org.ricks.net.http.logging.Logger;
import org.ricks.net.http.logging.LoggerFactory;
import org.ricks.net.http.utils.Constant;
import org.ricks.net.http.utils.NumberUtils;
import org.ricks.net.http.utils.WebSocketUtil;
import org.ricks.net.transport.AioQuickClient;
import org.ricks.net.transport.AioSession;
import org.ricks.net.util.StringUtils;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.charset.StandardCharsets;
import java.security.SecureRandom;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Consumer;

public class WebSocketClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketClient.class);

    private final WebSocketConfiguration configuration;

    /**
     * Header: Host
     */
    private final String hostHeader;
    /**
     * 客户端Client
     */
    private AioQuickClient client;

    /**
     * 绑定线程池资源组
     */
    private AsynchronousChannelGroup asynchronousChannelGroup;

    private boolean connected;

    private boolean firstConnected = true;

    /**
     * 消息处理器
     */
    private final HttpMessageProcessor processor = new HttpMessageProcessor();

    private final String uri;
    private WebSocketRequestImpl request;

    private static final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);


    public static void main(String[] args) throws IOException, InterruptedException {

        Map<String, WebSocketClient> clientMap = new HashMap<>();

        for (int i = 1; i <= 5 ; i++) {
            WebSocketClient client = new WebSocketClient("ws://localhost:7001/websocket");
            client.configuration().debug(true);
            int finalI = i;
            Thread.sleep(10);
            client.connect(new WebSocketListener() {
                @Override
                public void onOpen(WebSocketClient client, WebSocketResponse session) {
                    try {
                        client.sendBinary("ricks".getBytes());
////                        client.sendMessage("hello");
////                        Login.LoginRequest.Builder loginReq = Login.LoginRequest.newBuilder();
////
////                        loginReq.setUsername(  "ricks22" + finalI);
////                        loginReq.setDid("什么鬼");
////                        loginReq.setRpcId(1);
////                        ByteBuffer byteBuffer = ByteBuffer.allocate(2 + loginReq.build().toByteArray().length);
////                        byteBuffer.putShort((short) 100);
////                        byteBuffer.put(loginReq.build().toByteArray());
////                        byte[] a  = byteBuffer.array();;
////                        client.sendBinary(a);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }

                @Override
                public void onMessage(WebSocketClient client, String message) {
//                    System.out.println(message);
//                    LOGGER.info("msg, info:{}", message);
                }


                @Override
                public void onMessage(WebSocketClient client, byte[] message)  {
                    try {
                        ByteBuffer byteBuffer = ByteBuffer.wrap(message);
                        if(byteBuffer.getShort() == 101) {
                            byte[] a = new byte[byteBuffer.remaining()];
                            byteBuffer.get(a);
//                            Login.LoginResponse loginResponse = Login.LoginResponse.parseFrom(a);
//                            log.info("login success playerId:{} - name:{} ",
//                                    loginResponse.getPlayerInfo().getAccount(), loginResponse.getPlayerInfo().getName() );


//                            scheduledExecutorService.scheduleAtFixedRate(() -> {
//                                Chat.ChatReq.Builder builder = Chat.ChatReq.newBuilder();
//                                builder.setRpcId(1);
//                                String chatInfo = loginResponse.getPlayerInfo().getName() + ": 你好，我有一个帽衫！ --- " + finalI1;
//                                builder.setChatInfo(ByteString.copyFrom(chatInfo.getBytes()));
//
//                                ByteBuffer byteBuffer1 = ByteBuffer.allocate(2 + builder.build().toByteArray().length);
//                                byteBuffer1.putShort((short) 1171);
//                                byteBuffer1.put(builder.build().toByteArray());
//                                byte[] aa  = byteBuffer1.array();;
//                                try {
//                                    client.sendBinary(aa);
//                                } catch (IOException e) {
//                                    throw new RuntimeException(e);
//                                }
//                            }, 1000,5000,TimeUnit.MILLISECONDS);


//                            Recharge.CreateRechargeOrderReq.Builder builder =
//                                    Recharge.CreateRechargeOrderReq.newBuilder();
//                            builder.setRpcId(1);
//                            builder.setRechargeId(1);
//                                ByteBuffer byteBuffer1 = ByteBuffer.allocate(2 + builder.build().toByteArray().length);
//                                byteBuffer1.putShort((short) 1180);
//                                byteBuffer1.put(builder.build().toByteArray());
//                                byte[] aa  = byteBuffer1.array();;
//                                client.sendBinary(aa);

//                            for (int i = 0; i < 100; i++) {
//                                Chat.ChatReq.Builder builder = Chat.ChatReq.newBuilder();
//                                builder.setRpcId(1);
//                                String chatInfo = loginResponse.getPlayerInfo().getName() + ": 你好，我有一个帽衫！ --- " + i;
//                                builder.setChatInfo(ByteString.copyFrom(chatInfo.getBytes()));
//
//                                ByteBuffer byteBuffer1 = ByteBuffer.allocate(2 + builder.build().toByteArray().length);
//                                byteBuffer1.putShort((short) 1171);
//                                byteBuffer1.put(builder.build().toByteArray());
//                                byte[] aa  = byteBuffer1.array();;
//                                client.sendBinary(aa);
//                            }

//                            clientMap.put("loginResponse.getPlayerInfo().getAccount()", client);



//                                ByteBuffer byteBuffer1 = ByteBuffer.allocate(2);
//                                byteBuffer1.putShort((short) 1);
//                                byte[] aa  = byteBuffer1.array();;
//                                client.sendBinary(aa);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
            });
            Thread.sleep(10);
        }
//        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable() {
//            @Override
//            public void run() {
//                try {
//                    client.sendMessage("aaa" + System.currentTimeMillis());
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            }
//        }, 1, 1, TimeUnit.SECONDS);

//        Login.LoginRequest.Builder loginReq = Login.LoginRequest.newBuilder();
//        loginReq.setUsername("rickstestno5gm");
//        loginReq.setDid("什么鬼");
//        loginReq.setRpcId(1);
//        ByteBuffer byteBuffer = ByteBuffer.allocate(2 + loginReq.build().toByteArray().length);
//        byteBuffer.putShort((short) 100);
//        byteBuffer.put(loginReq.build().toByteArray());
//        byte[] a  = byteBuffer.array();;
//        client.sendBinary(a);
//
//
//        Thread.sleep(5000L);
//
//        Active.AccumulateLoginReq.Builder req = Active.AccumulateLoginReq.newBuilder();
//        req.setRpcId(1);
//        byte[] bytes = req.build().toByteArray();
//        byteBuffer = ByteBuffer.allocate(2 + bytes.length);
//        byteBuffer.putShort((short) 1060);
//        byteBuffer.put(bytes);
//        a  = byteBuffer.array();;
//        client.sendBinary(a);
//
//        Active.ClaimAccumulateLoginReq.Builder builder = Active.ClaimAccumulateLoginReq.newBuilder();
//        builder.setRpcId(1);
//        byteBuffer = ByteBuffer.allocate(2 + builder.build().toByteArray().length);
//        byteBuffer.putShort((short) 1061);
//        byteBuffer.put( builder.build().toByteArray());
//        byte[] aa  = byteBuffer.array();;
//
//        for (int i = 0; i < 100; i++) {
//            new Thread(() -> {
//                try {
//                    client.sendBinary(aa);
//                } catch (IOException e) {
//                    throw new RuntimeException(e);
//                }
//            }).start();
//        }


//        byteBuffer = ByteBuffer.allocate(2);
//        byteBuffer.putShort((short) 1);
//        byte[] aa  = byteBuffer.array();;
//        client.sendBinary(aa);

//        for (int i = 0; i < 10; i++) {
//            new Thread(() -> {
//                try {
//                    for (int j = 0; j < 100 ; j++) {
//                        client.sendBinary(aa);
//                    }
//                } catch (IOException e) {
//                    throw new RuntimeException(e);
//                }
//            }).start();
//        }

//        for (Map.Entry<String, WebSocketClient> entry: clientMap.entrySet()) {
//            for (int i = 0; i < 100; i++) {
//                Chat.ChatReq.Builder builder = Chat.ChatReq.newBuilder();
//                builder.setRpcId(1);
//                String chatInfo = entry.getKey() + ": 你好，我有一个帽衫！ --- " + i;
//                builder.setChatInfo(ByteString.copyFrom(chatInfo.getBytes()));
//
//                ByteBuffer byteBuffer = ByteBuffer.allocate(2 + builder.build().toByteArray().length);
//                byteBuffer.putShort((short) 1171);
//                byteBuffer.put(builder.build().toByteArray());
//                byte[] aa  = byteBuffer.array();;
//                entry.getValue().sendBinary(aa);
//            }
//
//
//        }
//        for (int i = 0; i < 100 ; i++) {
//            Common.BaseInfo.Builder builder1 = Common.BaseInfo.newBuilder();
//            builder1.setPlayerId("ricks-job-is-demo" + i);
//            builder1.setName("ricks-job-is-demo" + i);
//            builder1.setLv(1);
//            builder1.setHeadId(1);
//            builder1.setServerId(5);
//            builder1.setCombatNum(100);
//            baseInfoMap.put(builder1.getPlayerId(),builder1.build());
//        }

    }



    public WebSocketClient(String url) {
        int schemaIndex = url.indexOf("://");
        if (schemaIndex == -1) {
            throw new IllegalArgumentException("invalid url:" + url);
        }
        String schema = url.substring(0, schemaIndex);
        int uriIndex = url.indexOf("/", schemaIndex + 3);
        int portIndex = url.indexOf(":", schemaIndex + 3);
        boolean ws = Constant.SCHEMA_WS.equals(schema);
        boolean wss = !ws && Constant.SCHEMA_WSS.equals(schema);

        if (!ws && !wss) {
            throw new IllegalArgumentException("invalid url:" + url);
        }
        String host;
        int port;
        if (portIndex > 0) {
            host = url.substring(schemaIndex + 3, portIndex);
            port = NumberUtils.toInt(uriIndex > 0 ? url.substring(portIndex + 1, uriIndex) : url.substring(portIndex + 1), -1);
        } else if (uriIndex > 0) {
            host = url.substring(schemaIndex + 3, uriIndex);
            port = wss ? 443 : 80;
        } else {
            host = url.substring(schemaIndex + 3);
            port = wss ? 443 : 80;
        }
        if (port == -1) {
            throw new IllegalArgumentException("invalid url:" + url);
        }
        this.configuration = new WebSocketConfiguration(host, port);
        configuration.setWss(wss);
        hostHeader = configuration.getHost() + ":" + configuration.getPort();
        this.uri = uriIndex > 0 ? url.substring(uriIndex) : "/";

    }

    public WebSocketConfiguration configuration() {
        return configuration;
    }

    public void connect(WebSocketListener listener) throws IOException {
        if (connected) {
            AioSession session = client.getSession();
            if (session == null || session.isInvalid()) {
                close();
                connect(listener);
            }
            return;
        }

        try {
            if (firstConnected) {
                boolean noneSslPlugin = true;
                for (Plugin responsePlugin : configuration.getPlugins()) {
                    processor.addPlugin(responsePlugin);
                    if (responsePlugin instanceof SslPlugin) {
                        noneSslPlugin = false;
                    }
                }
                if (noneSslPlugin && configuration.isWss()) {
                    processor.addPlugin(new SslPlugin<>(new ClientSSLContextFactory()));
                }

                firstConnected = false;
            }
            connected = true;
            client = configuration.getProxy() == null ? new AioQuickClient(configuration.getHost(), configuration.getPort(), processor, processor) : new AioQuickClient(configuration.getProxy().getProxyHost(), configuration.getProxy().getProxyPort(), processor, processor);
            client.setBufferPagePool(configuration.getReadBufferPool(), configuration.getWriteBufferPool()).setReadBufferSize(configuration.readBufferSize());
            if (configuration.getConnectTimeout() > 0) {
                client.connectTimeout(configuration.getConnectTimeout());
            }
            if (asynchronousChannelGroup == null) {
                client.start();
            } else {
                client.start(asynchronousChannelGroup);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        AioSession session = client.getSession();
        DecoderUnit attachment = session.getAttachment();
        CompletableFuture<WebSocketResponseImpl> completableFuture = new CompletableFuture<>();
        completableFuture.thenAccept(new Consumer<WebSocketResponseImpl>() {
            @Override
            public void accept(WebSocketResponseImpl webSocketResponse) {
                try {
                    switch (webSocketResponse.getFrameOpcode()) {
                        case WebSocketUtil.OPCODE_TEXT:
                            listener.onMessage(WebSocketClient.this, new String(webSocketResponse.getPayload(), StandardCharsets.UTF_8));
                            break;
                        case WebSocketUtil.OPCODE_BINARY:
                            listener.onMessage(WebSocketClient.this, webSocketResponse.getPayload());
                            break;
                        case WebSocketUtil.OPCODE_CLOSE:
                            try {
                                listener.onClose(WebSocketClient.this, webSocketResponse, new CloseReason(webSocketResponse.getPayload()));
                            } finally {
                                WebSocketClient.this.close();
                            }
                            break;
                        case WebSocketUtil.OPCODE_PING:
                            System.out.println("ping...");
                            WebSocketUtil.send(request.getOutputStream(), WebSocketUtil.OPCODE_PONG, webSocketResponse.getPayload(), 0, webSocketResponse.getPayload().length);
//                        webSocketResponse.pong(webSocketResponse.getPayload());
                            break;
                        case WebSocketUtil.OPCODE_PONG:
//                                handlePong(request, response);
                            break;
                        case WebSocketUtil.OPCODE_CONTINUE:
                            LOGGER.warn("unSupport OPCODE_CONTINUE now,ignore payload: {}", StringUtils.toHexString(webSocketResponse.getPayload()));
                            break;
                        default:
                            throw new UnsupportedOperationException();
                    }
                } catch (Throwable throwable) {
                    listener.onError(WebSocketClient.this, webSocketResponse, throwable);
//                throw throwable;
                } finally {
                    CompletableFuture<WebSocketResponseImpl> completableFuture = new CompletableFuture<>();
                    completableFuture.thenAccept(this);
                    webSocketResponse.setFuture(completableFuture);
                    webSocketResponse.reset();
                    attachment.setState(DecoderUnit.STATE_BODY);
                }
            }
        });
        WebSocketResponseImpl webSocketResponse = new WebSocketResponseImpl(session, completableFuture);
        webSocketResponse.setResponseHandler(new ResponseHandler() {
            @Override
            public void onHeaderComplete(AbstractResponse abstractResponse) throws IOException {
                WebSocketResponseImpl webSocketResponse = (WebSocketResponseImpl) abstractResponse;
                super.onHeaderComplete(webSocketResponse);
                if (webSocketResponse.getStatus() != HttpStatus.SWITCHING_PROTOCOLS.value()) {
                    listener.onClose(WebSocketClient.this, webSocketResponse, new CloseReason(CloseReason.WRONG_CODE, ""));
                    return;
                }
                listener.onOpen(WebSocketClient.this, webSocketResponse);
            }

            @Override
            public void onBodyStream(ByteBuffer buffer, AbstractResponse abstractResponse) {
                WebSocketResponseImpl webSocketResponse = (WebSocketResponseImpl) abstractResponse;
                Decoder decoder = webSocketResponse.getDecoder().decode(buffer, webSocketResponse);
                webSocketResponse.setDecoder(decoder);
                if (decoder == WebSocket.PAYLOAD_FINISH) {
                    abstractResponse.getFuture().complete(abstractResponse);
                }
            }

        });

        attachment.setResponse(webSocketResponse);
        initRest();
    }

    private void initRest() throws IOException {
        request = new WebSocketRequestImpl(client.getSession());
        request.setUri(uri);
        request.setMethod(HttpMethodEnum.GET.getMethod());
        request.setProtocol(HttpProtocolEnum.HTTP_11.getProtocol());
        request.addHeader(HeaderNameEnum.HOST.getName(), hostHeader);
        request.addHeader(HeaderNameEnum.UPGRADE.getName(), HeaderValueEnum.WEBSOCKET.getName());
        request.setHeader(HeaderNameEnum.CONNECTION.getName(), HeaderValueEnum.UPGRADE.getName());
        request.setHeader(HeaderNameEnum.Sec_WebSocket_Key.getName(), generateSecWebSocketKey());
        request.setHeader(HeaderNameEnum.Sec_WebSocket_Version.getName(), "13");
//        request.setHeader(HeaderNameEnum.Sec_WebSocket_Protocol.getName(), HeaderValueEnum.PERMESSAGE_DEFLATE.getName());
        request.getOutputStream().flush();
    }

    /**
     * 在客户端握手中的|Sec-WebSocket-Key|头字段包括一个 base64 编码的值，如果解码，长度是 16 字节。
     */
    private String generateSecWebSocketKey() {
        byte[] keyBytes = new byte[16];
        new SecureRandom().nextBytes(keyBytes);
        return Base64.getUrlEncoder().withoutPadding().encodeToString(keyBytes);
    }


    /**
     * 发送消息
     *
     * @param message 消息内容
     * @throws IOException 如果在发送消息过程中发生I/O错误
     */
    public void sendMessage(String message) throws IOException {
        // 发送消息到服务器
        byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
        WebSocketUtil.sendMask(request.getOutputStream(), WebSocketUtil.OPCODE_TEXT, bytes, 0, bytes.length);
        request.getOutputStream().flush();
    }

    /**
     * 发送二进制消息
     *
     * @param bytes 二进制消息内容
     * @throws IOException 如果在发送过程中发生I/O错误
     */
    public void sendBinary(byte[] bytes) throws IOException {
        // 发送二进制消息到服务器
        WebSocketUtil.sendMask(request.getOutputStream(), WebSocketUtil.OPCODE_BINARY, bytes, 0, bytes.length);
        request.getOutputStream().flush();
    }


    public void setAsynchronousChannelGroup(AsynchronousChannelGroup asynchronousChannelGroup) {
        this.asynchronousChannelGroup = asynchronousChannelGroup;
    }

    public void close() {
        try {
            WebSocketUtil.sendMask(request.getOutputStream(), WebSocketUtil.OPCODE_CLOSE, new CloseReason(CloseReason.NORMAL_CLOSURE, "").toBytes());
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            connected = false;
            client.shutdownNow();
        }
    }

}
